package advertising.handler

import org.apache.commons.lang3.StringUtils
import org.apache.http.impl.client.HttpClients
import org.apache.spark.sql.Row

import scala.collection.mutable.ArrayBuffer

//  .--,       .--,
// ( (  \.---./  ) )
//  '.__/o   o\__.'
//     {=  ^  =}
//      >  -  <
//     /       \
//    //       \\
//   //|   .   |\\
//   "'\       /'"_.-~^`'-.
//      \  _  /--'         `
//    ___)( )(___
//   (((__) (__)))    高山仰止, 景行行止; 虽不能至, 心向往之.
//                    High mountains, scenery. Although not to, the heart yearns for it. 
object TagsHandler {

  /**
    * 关键字标签
    * @param row
    * @return Array[(String, Int)]((关键字, 数量))
    */
  def row2KeywordsTags(row: Row) = {
    val keywords = FieldValueHandler.getKeywords(row)
    if(StringUtils.isBlank(keywords)){
      Array[(String, Int)]()
    } else {
      val keywordsArr = StringUtils.split(keywords, "|")
      if(keywordsArr.length >= 3 && keywordsArr.length <= 8){
        val result = ArrayBuffer[(String, Int)]()
        for(keyword <- keywordsArr){
          result += ((keyword, 1))
        }
        result.toArray
      } else {
        Array[(String, Int)]()
      }
    }
  }

  /**
    * 设备标签
    * @param row
    * @return 系统、网络、运营商标签
    */
  def row2EquipmentTags(row: Row) = {
      val osName = FieldValueHandler.getOSName(row)
      val networkName = FieldValueHandler.getNetworkName(row)
      val ispName = FieldValueHandler.getIspName(row)
      Array[(String, Int)]((osName, 1), (networkName, 1), (ispName, 1))
  }

  /**
    * 渠道标签
    * @param row
    * @return 广告渠道标签
    */
  def row2ChannelTags(row: Row) = {
      val adPlatformProviderId = FieldValueHandler.getAdPlatformProviderId(row).toString
      Array[(String, Int)]((adPlatformProviderId, 1))
  }

  /**
    * 广告类型标签
    * @param row
    * @return 广告类型标签
    */
  def row2AdTags(row: Row) = {
    val adSpaceType = FieldValueHandler.getAdSpaceType(row)
    Array[(String, Int)]((adSpaceType, 1))
  }

  /**
    * 应用名称标签
    * @param row
    * @return 应用名称标签
    */
  def row2AppNameTags(row:Row) = {
    val appName = FieldValueHandler.getAppName(row)
    Array[(String, Int)]((appName, 1))
  }

  private def getBusinessTags(business: String) = {
    if(StringUtils.isBlank(business)) {
      Array[(String,Int)]()
    } else {
      val businessArr = StringUtils.split(business, ",")
      val result = ArrayBuffer[(String, Int)]()
      for(business <- businessArr) {
        result += ((business, 1))
      }
      result.toArray
    }
  }

  /**
    * 商圈标签
    * @param row
    * @return 商圈标签
    */
  def row2BusinessTags(row: Row) = {
    val latitude = FieldValueHandler.getLatitude(row)
    val longitude = FieldValueHandler.getLongitude(row)
    val business = LBSHandler.getBusinessFromDB(latitude, longitude)
    if(business == null){
      // 商圈知识库中没有
      val httpClient = HttpClients.createDefault()
      val businessFromBaidu = LBSHandler.getBusinessFromBaidu(latitude, longitude, httpClient)
      val geoHash = LBSHandler.getGeoHash(latitude, longitude)
      // 增加到商圈知识库中
      LBSHandler.save2DB(geoHash, businessFromBaidu)
      getBusinessTags(businessFromBaidu)
    } else {
      getBusinessTags(business)
    }
  }

  /**
    * row => 用户所有标签
    * @param row
    * @return 用户所有标签
    */
  def row2AllTags(row:Row):Array[(String, Int)] = {
    // 广告标签
    val adTags = row2AdTags(row)
    // 渠道标签
    val channelTags = row2ChannelTags(row)
    // 设备标签
    val equipmentTags = row2EquipmentTags(row)
    // 关键字标签
    val keywordsTags = row2KeywordsTags(row)
    // AppName 标签
    val appNameTags = row2AppNameTags(row)
    // 商圈标签
    val businessTags = row2BusinessTags(row)
    // 合并标签
    adTags ++ channelTags ++ equipmentTags ++ keywordsTags ++ appNameTags ++ businessTags
  }

}
